package util

import (
	"common/logging"
	"iot/iot_app_server/proto"
	"time"

	"github.com/Shopify/sarama"
	"github.com/bsm/sarama-cluster"
	"github.com/golang/protobuf/proto"
)

var (
	BroadcastConsumer *cluster.Consumer
	DispatchConsumer  *cluster.Consumer
	bStopConsumer     bool
	stopChan          = make(chan bool)
)

const (
	bConsumerGroupPrefix = "UPSTREAM_GROUP_BROADCAST_"
	dConsumerGroupPrefix = "UPSTREAM_GROUP_DISPATCH"
)

type UpstreamDispatcher func([]byte) error

func InitConsumer(dispatchFunc UpstreamDispatcher) error {
	config := cluster.NewConfig()
	var err error
	BroadcastConsumer, err = cluster.NewConsumer(Cfg.Server.KafkaBrokers, bConsumerGroupPrefix+Cfg.Server.ServerId, []string{Cfg.Server.UpstreamTopicBroadcast}, config)
	if err != nil {
		logging.Error("InitConsumer BroadcastConsumer failed,err=%s", err.Error())
		return err
	}

	DispatchConsumer, err = cluster.NewConsumer(Cfg.Server.KafkaBrokers, dConsumerGroupPrefix, []string{Cfg.Server.UpstreamTopicDispatch}, config)
	if err != nil {
		logging.Error("InitConsumer DispatchConsumer failed,err=%s", err.Error())
		return err
	}

	msgHandle := func(m *sarama.ConsumerMessage) {
		upMsg := &MSGBUS_DOWNSTREAM.MqttUpstreamMsg{}
		err := proto.Unmarshal(m.Value, upMsg)
		if err != nil {
			logging.Error("UpstreamDispatcher Unmarshal MqttUpstreamMsg failed")
			return
		}
		if upMsg.GetType() != MSGBUS_DOWNSTREAM.MsgType_msg_type_mqtt_pub {
			logging.Error("UpstreamDispatcher MqttUpstreamMsg invalid msg type")
			return
		}

		go func() {
			dispatchFunc(upMsg.GetBody())
		}()
	}

	go func() {
		for {
			timer := time.NewTimer(time.Second * time.Duration(1))
			var m *sarama.ConsumerMessage
			select {
			case m = <-BroadcastConsumer.Messages():
				if m == nil {
					logging.Error("BroadcastConsumer  get nil msg")
					//stopChan <- true
					//return
					continue
				}
				BroadcastConsumer.MarkOffset(m, "PullMessage")
				//logging.Debug("BroadcastConsumer get msg :%s", string(m.Value))
				msgHandle(m)
			case m = <-DispatchConsumer.Messages():
				if m == nil {
					logging.Error("DispatchConsumer get nil msg")
					//stopChan <- true
					//return
					continue
				}
				DispatchConsumer.MarkOffset(m, "PullMessage")
				//logging.Debug("DispatchConsumer get msg :%s", string(m.Value))
				msgHandle(m)
			case <-timer.C:
				if bStopConsumer {
					stopChan <- true
					return
				}
			}
		}
	}()
	return nil
}

//应该用waitgroup的,先弄个粗糙的
func StopConsumer() {
	bStopConsumer = true
	<-stopChan
	if BroadcastConsumer != nil {
		BroadcastConsumer.Close()
	}
	if DispatchConsumer != nil {
		DispatchConsumer.Close()
	}
}
